#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <ctype.h>

#include <time.h>

#include "sockets.h"
#include "netdb.h"
#include "def.h"
#include "uart.h"
#include "debug.h"
#include "os_api.h"
#include "dev_api.h"
#include "ps_api.h"
#include "file_api.h"

#include "dtu_socket.h"

#if 0
#define dtu_socket_log_debug(fmt, args...) do { sdklogConfig(1); LOG_PRINTF("dtu_socket: "fmt, ##args); } while(0)
#define dtu_socket_log_info(fmt, args...) do { sdklogConfig(1); LOG_PRINTF("dtu_socket: "fmt, ##args); } while(0)
#define dtu_socket_log_err(fmt, args...) do { sdklogConfig(1); LOG_PRINTF("dtu_socket: "fmt, ##args); } while(0)
#else
#define dtu_socket_log_debug(fmt, args...) LOG_PRINTF("dtu_socket: "fmt, ##args)
#define dtu_socket_log_info(fmt, args...) LOG_PRINTF("dtu_socket: "fmt, ##args)
#define dtu_socket_log_err(fmt, args...) LOG_PRINTF("dtu_socket: "fmt, ##args)
#endif

extern int check_network_ready(void);

#define DTU_SOCKET_MESSAGE_Q_SIZE 	sizeof(dtu_socket_msg_t)
#define DTU_SOCKET_MESSAGE_Q_MAX 	50

#define DTU_SOCKET_TASK_STACK_SIZE     (1024*5)
#define DTU_SOCKET_TASK_PRIORITY        86

#define DTU_SOCKET_TASK_NET_STACK_SIZE     (1024*5)
#define DTU_SOCKET_TASK_NET_PRIORITY        85

#define DTU_SOCKET_CACHE_MESSAGE_Q_SIZE 	sizeof(dtu_socket_msg_t)
#define DTU_SOCKET_CACHE_MESSAGE_Q_MAX 		300

#define DTU_SOCKET_TASK_CACHE_STACK_SIZE     (1024*5)
#define DTU_SOCKET_TASK_CACHE_PRIORITY        84

#define MAX_LINK_ID	2

#define MAX_SOCKET_CACHE_SIZE	(1024*50)

typedef struct dtu_socket_msg_s {
	uint32_t msgId;
	uint32_t extra;
	uint32_t len;
	void *data;
} dtu_socket_msg_t;

enum  {
	DTU_SOCKET_CMD_START = 0,
	DTU_SOCKET_CMD_STOP,
	DTU_SOCKET_CMD_SUSPEND ,
	DTU_SOCKET_CMD_SEND,

	DTU_SOCKET_CMD_MAX
};

typedef struct dtu_socket_data_s {
	
	OSMsgQRef cmdq;
	OSTaskRef task_ref;
	void *task_stack;

	void *task_net_stack;
	OSTaskRef task_net_ref;
	OSAFlagRef flag_net_ref;

	OSMsgQRef cacheq;
	void *task_cache_stack;
	OSTaskRef task_cache_ref;
	OSAFlagRef flag_cache_ref;	

	UINT32 cacheq_size;

	OSMutexRef mutex_ref;

	int inited;

	int link_id;

	int fd_sock;

	int stop_count;

	int connected;

    int heart_miss_count;

	UINT32 task_timeout;

	socket_info_t *sockcfg;
} dtu_socket_data_t;


static dtu_socket_data_t *dtu_socket_ad[MAX_LINK_ID] = {NULL, NULL};


static socket_info_t local_socket_info[MAX_LINK_ID] = {0};

static char *sockcfg_nvm_file[MAX_LINK_ID] = {"sockcfg_nvm_1.info", "sockcfg_nvm_2.info"};

void dtu_socket_cfg_init(int link_id)
{
	FILE_ID fd;

	socket_info_t *sockcfg = &local_socket_info[link_id];
	
	memset(sockcfg, 0x00, sizeof(socket_info_t));
	
	fd = FDI_fopen(sockcfg_nvm_file[link_id], "rb");

	dtu_socket_log_info("read file %s, fd %d!\n", sockcfg_nvm_file[link_id], fd);	
	
	if (fd != 0) {
		FDI_fseek(fd, 0, SEEK_SET);
		FDI_fread(sockcfg, sizeof(socket_info_t), 1, fd);
		FDI_fclose(fd);
	} else {
		sockcfg->keepalive = 1;
		sockcfg->keepidle= 7200;
		sockcfg->keepinterval = 75;
		sockcfg->keepcount = 9;

		sockcfg->connect_retry_count= 3;
		sockcfg->send_retry_count= 3;
	}
	
}

socket_info_t *dtu_socket_cfg_get(int link_id)
{
	return &local_socket_info[link_id];
}

void dtu_socket_cfg_update_nvm(int link_id)
{
	FILE_ID fd;

	socket_info_t *sockcfg = &local_socket_info[link_id];

	fd = FDI_fopen(sockcfg_nvm_file[link_id], "wb");

	dtu_socket_log_info("write file %s, fd %d!\n", sockcfg_nvm_file[link_id], fd);	
	
	if (fd != 0) {
		FDI_fseek(fd, 0, SEEK_SET);	
		FDI_fwrite(sockcfg, sizeof(socket_info_t), 1, fd);
		FDI_fclose(fd);
	}	
}

void dtu_socket_cfg_remove(int link_id)
{
	FDI_remove(sockcfg_nvm_file[link_id]);
}


int dtu_socket_status(int link_id)
{
	int ret = 0;
	dtu_socket_data_t *ad = dtu_socket_ad[link_id];

	if (ad && ad->connected == 1) {
		ret = 1;
	}

	dtu_socket_log_info("link_id %d, dtu_socket_status %d\n", ad->link_id, ret);

	return ret;
}

int dtu_socket_send_msg(uint32_t msgId, uint32_t extra, void *data)
{
	dtu_socket_data_t *ad = data;
	dtu_socket_msg_t msg;

	if (ad && ad->cmdq) {
		memset(&msg, 0x00, sizeof(msg));
		msg.msgId = msgId;
		msg.extra = extra;
		msg.data = NULL;
		if (OS_SUCCESS == OSAMsgQSend(ad->cmdq, sizeof(msg), (UINT8 *)&msg, OSA_NO_SUSPEND)) {
			return 0;
		}
	}
	
	return -1;
}


int dtu_socket_send_msg_from_uart(int link_id, char *data, int len)
{
	dtu_socket_data_t *ad = dtu_socket_ad[link_id];
	dtu_socket_msg_t msg;

	dtu_socket_log_info("link_id %d, cacheq_size %d\n", link_id, ad?ad->cacheq_size:0);

	if (ad && ad->cacheq && ad->cacheq_size < MAX_SOCKET_CACHE_SIZE) {
		memset(&msg, 0x00, sizeof(msg));
		msg.msgId = DTU_SOCKET_CMD_SEND;
		msg.extra = 0;
		msg.data = malloc(len+1);
		memset(msg.data, 0x00, (len+1));
		memcpy(msg.data, data, len);
		msg.len = len;

		ad->cacheq_size += len;
		if (OS_SUCCESS == OSAMsgQSend(ad->cacheq, sizeof(msg), (UINT8 *)&msg, OSA_NO_SUSPEND)) {
			return 0;
		}
	}

	dtu_socket_log_info("link_id %d, socket not ready\n", ad->link_id);

	return -1;
}


void send_socket_data_to_uart(int link_id, char *data, int len)
{
	dtu_socket_log_info("link_id %d, len %d\n", link_id, len);
	
	if (link_id == 0){
		hal_UartWrite(UART_PORT_2, (UINT8 *)data, len);
	} else if (link_id == 1){
		hal_UartWrite(UART_PORT_3, (UINT8 *)data, len);
	}
}

static void dtu_socket_net_close(dtu_socket_data_t *ad)
{
	int fd, closing = 0;

	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);
	
	OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);
	if (ad->fd_sock >= 0) {
		fd = ad->fd_sock;
		dtu_socket_log_info("%s: link_id %d, try close connection\n", __func__, ad->link_id);
		ad->fd_sock = -1;
		close(fd);
		closing = 1;
		dtu_socket_log_info("%s: link_id %d, close the connection\n", __func__, ad->link_id);		
	}
	OSAMutexUnlock(ad->mutex_ref);
	
	if (closing)
		OSATaskYield();
	
	dtu_socket_log_info("%s: exit link_id %d\n", __func__, ad->link_id);
}

static int dtu_socket_net_send(dtu_socket_data_t *ad, const void *data, size_t sz)
{
	const char *d = (const char *)data;
	size_t cur = 0;
	int ret;
	while (cur < sz) {
		ret = send(ad->fd_sock, d + cur, sz - cur, 0);
		if (ret < 0) {
			return -1;
		} else if (!ret) {
			return -1;
		}
		cur += ret;
	}

	dtu_socket_log_info("%s,%d: send %d DATA: %s\n", __func__, __LINE__, sz, d);

	return 0;
}

static int dtu_socket_net_init(dtu_socket_data_t *ad)
{
	int ret = -1;
	struct timeval tv;
	fd_set rfds;
	char buf[1204] = {0};

	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);

	if (ad->sockcfg->login_buf_len) {
		ret = dtu_socket_net_send(ad, ad->sockcfg->login_buf, ad->sockcfg->login_buf_len);
		if (ret < 0) {
			ret = -1;
			goto clean;
		}
	} 

	if (ad->sockcfg->login_resp_buf_len) {

		dtu_socket_log_info("%s,link_id %d: send login wait recv login_resp\n", __func__, ad->link_id);

		FD_ZERO(&rfds);
		FD_SET(ad->fd_sock, &rfds);

		if (ad->sockcfg->login_timeout > 0)
			tv.tv_sec = ad->sockcfg->login_timeout;
		else
			tv.tv_sec = 10;

		tv.tv_usec = 0;
		ret = select(ad->fd_sock + 1, &rfds, NULL, NULL, &tv);
		if (ret != 1 || !FD_ISSET(ad->fd_sock, &rfds)) {
			ret = -1;
			dtu_socket_log_info("%s,link_id %d: login failed!\n", __func__, ad->link_id);
			goto clean;
		}
		ret = recv(ad->fd_sock, buf, sizeof(buf)-1, 0);
		if (ret <= 0) {
			ret = -1;
			dtu_socket_log_info("%s,link_id %d: login failed!\n", __func__, ad->link_id);		
			goto clean;
		}
		buf[ret] = '\0';

		dtu_socket_log_info("%s,link_id %d: recv %d, data %s!\n", __func__, ad->link_id, ret, buf);		

		if (ret >= ad->sockcfg->login_resp_buf_len && !memcmp(buf, ad->sockcfg->login_resp_buf, ad->sockcfg->login_resp_buf_len)) {
			goto init_ok;
		}

		ret = -1;
		
		dtu_socket_log_info("%s,link_id %d: login failed!\n", __func__, ad->link_id);		
	}
		
	goto clean;

init_ok:

	ret = 0;
clean:

	return ret;
}

static int dtu_socket_net_connect(dtu_socket_data_t *ad, int fd, const struct sockaddr *addr, socklen_t addrlen)
{
	int ret;
	struct timeval tv;
	int sockflag, optval;
	socklen_t optlen;
	struct linger lin;
	fd_set wfds;

	sockflag = fcntl(fd, F_GETFL, 0);
	if (sockflag == -1) {
		ret = -1;
		goto clean;
	}
	ret = fcntl(fd, F_SETFL, sockflag | O_NONBLOCK);
	if (ret) {
		ret = -1;
		goto clean;
	}

	ret = connect(fd, addr, addrlen);
	if (!ret) {
		goto connect_ok;
	}
	if (lwip_getsockerrno(fd) != EINPROGRESS) {
		ret = -1;
		goto clean;
	}

	FD_ZERO(&wfds);
	FD_SET(fd, &wfds);
	tv.tv_sec = 10;
	tv.tv_usec = 0;
	ret = select(fd+1, NULL, &wfds, NULL, &tv);
	if (ret != 1 || !FD_ISSET(fd, &wfds)) {
		ret = -1;
		goto clean;
	}
	optlen = sizeof(optval);
	ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &optval, (socklen_t *)&optlen);
	if (ret || optval) {
		goto clean;
	}

connect_ok:
	sockflag = fcntl(fd, F_GETFL, 0);
	if (sockflag == -1) {
		ret = -1;
		goto clean;
	}
	ret = fcntl(fd, F_SETFL, sockflag & (~O_NONBLOCK));
	if (ret) {
		ret = -1;
		goto clean;
	}

	optval = 1;
	setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &optval, sizeof(optval));
	lin.l_onoff = 1;
	lin.l_linger = 0;
	setsockopt(fd, SOL_SOCKET, SO_LINGER, &lin, sizeof(lin));

	return 0;
clean:
	return ret;
}

static int dtu_socket_net_create(dtu_socket_data_t *ad)
{
	struct sockaddr_in addr;
	struct sockaddr_in6 addr6;
	struct addrinfo hints, *result = NULL, *rp;
	struct timeval tv;
	int fd = -1, ret = -1;
	char buf[64];
	
	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);	

	if (!ad->sockcfg->host[0] || !ad->sockcfg->port) {
		ret = -1;
		goto clean;
	}

	memset(&addr, 0x00, sizeof(addr));
	memset(&addr6, 0x00, sizeof(addr6));
	
	if (inet_pton(AF_INET, ad->sockcfg->host, &addr.sin_addr) == 1) {
		fd = socket(AF_INET, SOCK_STREAM, 0);
		if (fd < 0) {
			ret = -1;
			goto clean;
		}
		addr.sin_family = AF_INET;
		addr.sin_port = htons(ad->sockcfg->port);
		ret = dtu_socket_net_connect(ad, fd, (struct sockaddr *)&addr, sizeof(addr));
		if (ret < 0) {
			goto clean;
		}
	} else if (inet_pton(AF_INET6, ad->sockcfg->host, &addr6.sin6_addr) == 1) {
		fd = socket(AF_INET6, SOCK_STREAM, 0);
		if (fd < 0) {
			ret = -1;
			goto clean;
		}
		addr6.sin6_len = sizeof(addr6);
		addr6.sin6_family = AF_INET6;
		addr6.sin6_port = htons(ad->sockcfg->port);
		ret = dtu_socket_net_connect(ad, fd, (struct sockaddr *)&addr6, sizeof(addr6));
		if (ret < 0) {
			goto clean;
		}
	} else {
		snprintf(buf, sizeof(buf), "%u", ad->sockcfg->port);

		memset(&hints, 0, sizeof(hints));
		hints.ai_flags = 0;
		hints.ai_family = AF_UNSPEC;
		hints.ai_socktype = SOCK_STREAM;
		hints.ai_protocol = 0;
		ret = getaddrinfo(ad->sockcfg->host, buf, &hints, &result);
		if (ret) {
			ret = -1;
			goto clean;
		}
		for (rp = result; rp; rp = rp->ai_next) {
			fd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
			if (fd < 0) {
				continue;
			}
			ret = dtu_socket_net_connect(ad, fd, rp->ai_addr, rp->ai_addrlen);
			if (ret < 0) {
				close(fd);
				fd = -1;
				continue;
			}
			break;
		}
		freeaddrinfo(result);
		result = NULL;

		if (fd < 0) {
			ret = -1;
			goto clean;
		}
	}

	ad->fd_sock = fd;
	ret = 0;

clean:
	if (result)
		freeaddrinfo(result);
	
	if (ret < 0 && fd >= 0)
		close(fd);
	
	dtu_socket_log_info("%s: exit link_id %d, ret %d\n", __func__, ad->link_id, ret);	

	return ret;
}

static int dtu_socket_net_recv(dtu_socket_data_t *ad)
{
	int fd, ret, closing = 0;
	uint16_t sz;
	struct timeval tv;
	fd_set rfds;
	char buf[2048] = {0};

	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);

	OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);
	if (ad->fd_sock < 0) {
		dtu_socket_log_info("%s: link_id %d, sock has closed\n", __func__, ad->link_id);
		ret = -1;
		closing = 1;
		goto net_err;
	}
	fd = ad->fd_sock;
	OSAMutexUnlock(ad->mutex_ref);

	dtu_socket_log_info("%s: link_id %d, waiting...\n", __func__, ad->link_id);	
	FD_ZERO(&rfds);
	FD_SET(fd, &rfds);
	ret = select(fd + 1, &rfds, NULL, NULL, NULL);
	if (ret <= 0 || !FD_ISSET(fd, &rfds)) {
		dtu_socket_log_info("%s: link_id %d, select error: %d\n", __func__, ad->link_id, ret);	
		OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);
		ret = -1;
		if (ad->fd_sock < 0) {
			dtu_socket_log_info("%s: link_id %d, sock is closing when waiting\n", __func__, ad->link_id);				
			closing = 1;
		}
		goto net_err;
	}
	
	OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);	
	if (ad->fd_sock < 0) {
		dtu_socket_log_info("%s: link_id %d, sock is closing when waiting\n", __func__, ad->link_id);				
		ret = -1;
		closing = 1;
		goto net_err;
	}
	ret = ioctlsocket(ad->fd_sock, FIONREAD, &sz);
	if (ret < 0 || !sz) {
		dtu_socket_log_info("%s: link_id %d, data is empty\n", __func__, ad->link_id);		
		ret = -1;
		goto net_err;
	}
	ret = recv(ad->fd_sock, buf, sizeof(buf)-1, 0);
	if (ret <= 0) {
		dtu_socket_log_info("%s: link_id %d, recv error\n", __func__, ad->link_id);		
		ret = -1;
		goto net_err;
	}
	buf[ret] = '\0';

	dtu_socket_log_info("%s: link_id %d, recv %d data: %s\n", __func__, ad->link_id, ret, buf);	
	
	ad->heart_miss_count = 0;

	if (ad->sockcfg->heart_resp_buf_len 
			&& ret == ad->sockcfg->heart_resp_buf_len 
			&& !memcmp(buf, ad->sockcfg->heart_resp_buf, ad->sockcfg->heart_resp_buf_len)) {
        dtu_socket_log_info("%s: recv heart response\n", __func__);
        ret = 0;
        goto clean;
    } 

	send_socket_data_to_uart(ad->link_id, buf, ret);
	ret = 0;
	goto clean;
net_err:
    if (ad->fd_sock >= 0) {
        close(ad->fd_sock);
        ad->fd_sock = -1;
	
		dtu_socket_log_info("%s: link_id %d, close the connection\n", __func__, ad->link_id);							
    }
    if (!closing) {
        ad->task_timeout = OSA_SUSPEND;
        dtu_socket_send_msg(DTU_SOCKET_CMD_STOP, 0, ad);
    }

clean:
	OSAMutexUnlock(ad->mutex_ref);
	
	dtu_socket_log_info("%s: exit link_id %d\n", __func__, ad->link_id);

	return ret;
}

static void dtu_socket_timeout(dtu_socket_data_t *ad)
{
	int ret, i;

	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);
	
	OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);
	
	if (ad->fd_sock < 0) {
		dtu_socket_log_info("%s: link_id %d, sock closed\n", __func__, ad->link_id);
		ad->task_timeout = OSA_SUSPEND;
		goto clean;
	}

	for (i = 0; i <= ad->sockcfg->send_retry_count; i++) {
		ret = dtu_socket_net_send(ad, ad->sockcfg->heart_buf, ad->sockcfg->heart_buf_len);
		if (ret < 0) {
			OSATaskSleep(5*20); // 0.5s
			continue;
		}
		dtu_socket_log_info("%s: link_id %d, heart send ok\n", __func__, ad->link_id);
		break;
	}

	if (i > ad->sockcfg->send_retry_count) {

		dtu_socket_log_info("%s: link_id %d, heart send error\n", __func__, ad->link_id);				
		
		ad->task_timeout = OSA_SUSPEND;
		
		dtu_socket_send_msg(DTU_SOCKET_CMD_STOP, 0, ad);
		
		goto clean;
	}
	
	ad->heart_miss_count ++;
	
	if (ad->sockcfg->heart_interval > 0)
		ad->task_timeout = ad->sockcfg->heart_interval * 200;
	else
		ad->task_timeout = OSA_SUSPEND;

clean:
	OSAMutexUnlock(ad->mutex_ref);
	
	dtu_socket_log_info("%s: exit link_id %d\n", __func__, ad->link_id);
}

static void dtu_socket_start(dtu_socket_data_t *ad, void* data)
{
	int ret, i, fd;

	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);
	
	dtu_socket_log_info("%s: link_id %d, host %s:%d\n", __func__, ad->link_id, ad->sockcfg->host, ad->sockcfg->port);

	dtu_socket_log_info("%s: link_id %d, login %d, %s\n", __func__, ad->link_id, ad->sockcfg->login_buf_len, ad->sockcfg->login_buf);
	
	dtu_socket_log_info("%s: link_id %d, login_resp %d, %s\n", __func__, ad->link_id, ad->sockcfg->login_resp_buf_len, ad->sockcfg->login_resp_buf);

	dtu_socket_log_info("%s: link_id %d, heart_interval %d\n", __func__, ad->link_id, ad->sockcfg->heart_interval);

	dtu_socket_log_info("%s: link_id %d, heart %d, %s\n", __func__, ad->link_id, ad->sockcfg->heart_buf_len, ad->sockcfg->heart_buf);
	
	dtu_socket_log_info("%s: link_id %d, heart_resp %d, %s\n", __func__, ad->link_id, ad->sockcfg->heart_resp_buf_len, ad->sockcfg->heart_resp_buf);

	//dtu_socket_net_close(ad);
	ad->connected = 0;

	OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);

	for (i = 0; i <= ad->sockcfg->connect_retry_count; i++) {
		ret = dtu_socket_net_create(ad);
		if (ret < 0) {
			OSATaskSleep(5*20); // 0.5s
			continue;
		}
		break;
	}
	
	if (i > ad->sockcfg->connect_retry_count) {
		dtu_socket_log_info("%s: link_id %d, connect server failed\n", __func__, ad->link_id);		
		goto init_err;
	}

	dtu_socket_log_info("%s: link_id %d, connect server ok\n", __func__, ad->link_id);		

	if (!ad->sockcfg->heart_buf_len && ad->sockcfg->keepalive) {

		int err;
		int optval;
		
		optval = ad->sockcfg->keepalive;
		err = setsockopt(ad->fd_sock, SOL_SOCKET, SO_KEEPALIVE, (void *)&optval, sizeof(int));
		dtu_socket_log_info("%s: link_id %d, keepalive setsockopt result %d\n", __func__, ad->link_id, err);

		optval = ad->sockcfg->keepidle;
		err = setsockopt(ad->fd_sock, IPPROTO_TCP, TCP_KEEPIDLE, (void *)&optval, sizeof(int));
		dtu_socket_log_info("%s: link_id %d, keepidle setsockopt result %d\n", __func__, ad->link_id, err);

		optval = ad->sockcfg->keepinterval;		
		err = setsockopt(ad->fd_sock, IPPROTO_TCP, TCP_KEEPINTVL, (void *)&optval, sizeof(int));
		dtu_socket_log_info("%s: link_id %d, keepinterval setsockopt result %d\n", __func__, ad->link_id, err);
		
		optval = ad->sockcfg->keepcount;
		err = setsockopt(ad->fd_sock, IPPROTO_TCP, TCP_KEEPCNT, (void *)&optval, sizeof(int));
		dtu_socket_log_info("%s: link_id %d, keepcount setsockopt result %d\n", __func__, ad->link_id, err);
	}

	if ( ad->sockcfg->login_buf_len ) {
		ret = dtu_socket_net_init(ad);
		if (ret < 0) {
			goto init_err;
		} 
	} 	

	dtu_socket_log_info("%s: link_id %d, init ok\n", __func__, ad->link_id);

	ad->connected = 1;

	OSAFlagSet(ad->flag_net_ref, 1, OSA_FLAG_OR);

	if (!ad->sockcfg->heart_buf_len || ad->sockcfg->heart_interval == 0) {
		ad->task_timeout = OSA_SUSPEND;
	} else {
		if ( !ad->sockcfg->login_buf_len ) {
			ad->task_timeout = 1* 200;	 // send heart immediately, to make server immediately receive heart after connected.
		} else {
			ad->task_timeout = ad->sockcfg->heart_interval * 200;
		}
	} 
	
	dtu_socket_send_msg(DTU_SOCKET_CMD_SUSPEND, 0, ad);

	goto clean;

init_err:
	dtu_socket_log_info("%s: link_id %d, dtu_socket_send_msg stop cmd\n", __func__, ad->link_id);

	dtu_socket_send_msg(DTU_SOCKET_CMD_STOP, 0, ad);
	
	goto clean;

clean:
	OSAMutexUnlock(ad->mutex_ref);

	dtu_socket_log_info("%s: exit link_id %d\n", __func__, ad->link_id);	
}

static void dtu_socket_stop(dtu_socket_data_t *ad, void* data)
{
	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);	
	
	ad->connected = 0;

	dtu_socket_net_close(ad);
	
	OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);
	
	ad->task_timeout = OSA_SUSPEND;	
	
	OSAMutexUnlock(ad->mutex_ref);

	if (0 == check_network_ready()){
		int try = 100;

		ad->stop_count++;

		if (ad->stop_count >= 5){
			dev_CtrlReset(0);
		}

		while(try--){
			if (1 == check_network_ready()){
				break;
			}
			OSATaskSleep(200);
		}
	} else {

		ad->stop_count++;

		if (ad->stop_count >= 100){
			dev_CtrlReset(0);
		}

		OSATaskSleep(5*200);
	}

	dtu_socket_log_info("%s: exit link_id %d,send start\n", __func__, ad->link_id);	

	dtu_socket_send_msg(DTU_SOCKET_CMD_START, 0, ad);

	dtu_socket_log_info("%s: exit link_id %d\n", __func__, ad->link_id);	
}

static void dtu_socket_suspend(dtu_socket_data_t *ad, void* data)
{
	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);	
	
	OSAMutexLock(ad->mutex_ref, OSA_SUSPEND);
	
	dtu_socket_log_info("%s: link_id %d, suspending...\n", __func__, ad->link_id);	
	
	OSAMutexUnlock(ad->mutex_ref);
	
	dtu_socket_log_info("%s: exit link_id %d\n", __func__, ad->link_id);	
}

static void dtu_socket_send(dtu_socket_data_t *ad, void* data)
{
	int ret;
	dtu_socket_msg_t *msg = (dtu_socket_msg_t *)data;

	dtu_socket_log_info("%s: link_id %d\n", __func__, ad->link_id);	

	if (msg->data == NULL){
		dtu_socket_log_info("%s: link_id %d, msg->data is null\n", __func__, ad->link_id);	
		return;
	}
	
	ret = dtu_socket_net_send(ad, msg->data, msg->len);
	
	dtu_socket_log_info("%s: exit link_id %d, ret %d\n", __func__, ad->link_id, ret);	
}

static void dtu_socket_cache_task(void *param)
{
	int ret;
	UINT32 mqcount = 0;
	dtu_socket_msg_t msg;
	dtu_socket_data_t *ad = param;

	char name_string[32] = {0};
	sprintf(name_string, "ds_cacheq%d", ad->link_id);
	ret = OSAMsgQCreate(&ad->cacheq, name_string, DTU_SOCKET_CACHE_MESSAGE_Q_SIZE, DTU_SOCKET_CACHE_MESSAGE_Q_MAX, OS_FIFO);
	ASSERT(ret == OS_SUCCESS);	

	while (1) {
		ret = OSAMsgQRecv(ad->cacheq, (UINT8 *)&msg, DTU_SOCKET_CACHE_MESSAGE_Q_SIZE, OSA_SUSPEND);
		dtu_socket_log_info("%s: link_id: %d, OSAMsgQRecv ret: %d\n", __func__, ad->link_id, ret);
		if (ret == OS_SUCCESS) {
			while(ad->connected != 1){
				mqcount = 0;
				OSAMsgQPoll(ad->cacheq, &mqcount);
				dtu_socket_log_info("%s: link_id: %d, wait connected, mqcount %d!\n", __func__, ad->link_id, mqcount);
				OSATaskSleep(200);
			}

			ad->cacheq_size -= msg.len;
			// DTU_SOCKET_CACHE_MESSAGE_Q_SIZE == DTU_SOCKET_MESSAGE_Q_SIZE
			ret = OSAMsgQSend(ad->cmdq, DTU_SOCKET_MESSAGE_Q_SIZE, (UINT8 *)&msg, OSA_NO_SUSPEND);
			
			dtu_socket_log_info("%s: link_id: %d, cmdq send ret: %d, cacheq_size %d\n", __func__, ad->link_id, ret, ad->cacheq_size);		
		}
	}
}

static void dtu_socket_net_task(void *param)
{
	int ret;
	UINT32 flags;
	dtu_socket_data_t *ad = param;

	while (1) {
		ret = OSAFlagWait(ad->flag_net_ref, 0xffffffff, OSA_FLAG_OR_CLEAR, &flags, OSA_SUSPEND);
		ASSERT(ret == OS_SUCCESS);
		do {
			ret = dtu_socket_net_recv(ad);
			
			dtu_socket_log_info("%s: link_id %d, app_net_recv ret: %d\n", __func__, ad->link_id, ret);	
		} while (!ret);
	}
}

static void dtu_socket_task(void *param)
{
	int ret;
	UINT32 flag;	
	dtu_socket_msg_t msg;
	dtu_socket_data_t *ad = param;

	dtu_socket_log_info("%s: link_id: %d+\n", __func__, ad->link_id);
	while (1) {
		
		ret = OSAMsgQRecv(ad->cmdq, (UINT8 *)&msg, DTU_SOCKET_MESSAGE_Q_SIZE, ad->task_timeout);
		dtu_socket_log_info("%s: link_id: %d, OSAMsgQRecv ret: %d+\n", __func__, ad->link_id, ret);
		if (ret == OS_TIMEOUT) {
			
			dtu_socket_timeout(ad);
		} else if (ret == OS_SUCCESS) {
		
			dtu_socket_log_info("%s: link_id: %d, msg.msgId: %d+\n", __func__, ad->link_id, msg.msgId);
			switch (msg.msgId) {
				case DTU_SOCKET_CMD_START:
					dtu_socket_start(ad, &msg);
					break;
				case DTU_SOCKET_CMD_STOP:
					dtu_socket_stop(ad, &msg);
					break;
				case DTU_SOCKET_CMD_SUSPEND:
					dtu_socket_suspend(ad, &msg);
					break;
				case DTU_SOCKET_CMD_SEND:
					dtu_socket_send(ad, &msg);
					break;
				default:
					break;
			}

			if (msg.data){
				free(msg.data);
				msg.data = NULL;
			}			
		} else {
		
			ASSERT(0);
		}
	}
}

int dtu_socket_init(int link_id, socket_info_t *socket_ptr)
{
	int ret;

	char name_string[32] = {0};
	
	dtu_socket_data_t *ad;

	dtu_socket_log_info("%s: +\n", __func__);

	if (link_id > MAX_LINK_ID){
		dtu_socket_log_info("%s: link_id %d\n", __func__, link_id);	
		return -1;
	}

	ad = dtu_socket_ad[link_id] = malloc(sizeof(*dtu_socket_ad[link_id]));
	ASSERT(dtu_socket_ad[link_id]);
	memset(ad, 0x00, sizeof(*ad));

	ad->link_id = link_id;
	ad->fd_sock = -1;
	ad->task_timeout = OSA_SUSPEND;
	ad->sockcfg = socket_ptr;
	
	ret = OSAMutexCreate(&ad->mutex_ref, OS_PRIORITY);
	ASSERT(ret == OS_SUCCESS);

	ret = OSAFlagCreate(&ad->flag_net_ref);
	ASSERT(ret == OS_SUCCESS);

	ad->task_cache_stack = malloc(DTU_SOCKET_TASK_CACHE_STACK_SIZE);
	ASSERT(ad->task_cache_stack);
	sprintf(name_string, "ds_cache_task%d", link_id);
	ret = OSATaskCreate(&ad->task_cache_ref, ad->task_cache_stack, DTU_SOCKET_TASK_CACHE_STACK_SIZE, DTU_SOCKET_TASK_CACHE_PRIORITY, name_string, dtu_socket_cache_task, ad);
	ASSERT(ret == OS_SUCCESS);	
	
	ad->task_net_stack = malloc(DTU_SOCKET_TASK_NET_STACK_SIZE);
	ASSERT(ad->task_net_stack);
	sprintf(name_string, "ds_net_task%d", link_id);
	ret = OSATaskCreate(&ad->task_net_ref, ad->task_net_stack, DTU_SOCKET_TASK_NET_STACK_SIZE, DTU_SOCKET_TASK_NET_PRIORITY, name_string, dtu_socket_net_task, ad);
	ASSERT(ret == OS_SUCCESS);

	sprintf(name_string, "ds_msg%d", link_id);
	ret = OSAMsgQCreate(&ad->cmdq, name_string, DTU_SOCKET_MESSAGE_Q_SIZE, DTU_SOCKET_MESSAGE_Q_MAX, OS_FIFO);
	ASSERT(ret == OS_SUCCESS);
	
	ad->task_stack = malloc(DTU_SOCKET_TASK_STACK_SIZE);
	ASSERT(ad->task_stack);
	sprintf(name_string, "ds_task%d", link_id);
	ret = OSATaskCreate(&ad->task_ref, ad->task_stack, DTU_SOCKET_TASK_STACK_SIZE, DTU_SOCKET_TASK_PRIORITY, name_string, dtu_socket_task, ad);
	ASSERT(ret == OS_SUCCESS);

	ad->inited = 1;

	dtu_socket_send_msg(DTU_SOCKET_CMD_START, 0, ad);

	dtu_socket_log_info("%s: -\n", __func__);
	return 0;
}

void dtu_socket_app_start(void)
{	
	OS_STATUS status;

	socket_info_t *socket_ptr_0 = NULL;
	socket_info_t *socket_ptr_1 = NULL;

	dtu_socket_log_info("%s: starting...\n", __FUNCTION__);

	dtu_socket_cfg_init(0);
	dtu_socket_cfg_init(1);

	socket_ptr_0 = dtu_socket_cfg_get(0);
	socket_ptr_1 = dtu_socket_cfg_get(1);

	dtu_socket_log_info("%s: link_id 0, %s,%d\n", __func__, socket_ptr_0->host, socket_ptr_0->port);
	dtu_socket_log_info("%s: link_id 0, keepalive %d,%d,%d,%d\n", __func__, 
								socket_ptr_0->keepalive, 
								socket_ptr_0->keepidle,
								socket_ptr_0->keepinterval,
								socket_ptr_0->keepcount);
	if  (strlen(socket_ptr_0->host) && (socket_ptr_0->port >= 0)) {
		dtu_socket_init(0, socket_ptr_0);
	}

	dtu_socket_log_info("%s: link_id 1, %s,%d\n", __func__, socket_ptr_1->host, socket_ptr_1->port);
	dtu_socket_log_info("%s: link_id 1, keepalive %d,%d,%d,%d\n", __func__, 
								socket_ptr_1->keepalive, 
								socket_ptr_1->keepidle,
								socket_ptr_1->keepinterval,
								socket_ptr_1->keepcount);	
	if  (strlen(socket_ptr_1->host) && (socket_ptr_1->port >= 0)) {
		dtu_socket_init(1, socket_ptr_1);
	}

	dtu_socket_log_info("%s: exiting...\n", __FUNCTION__);
}
